package com.icbc.dccsh.reader;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.icbc.dccsh.logpumper.IConfigurable;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;

public class RedisReader extends Thread implements IConfigurable {
	public static final Logger LOG = LoggerFactory.getLogger(RedisReader.class);

	private LinkedBlockingQueue<String> queue;
//	private JedisPool _pool;
	private String _host;
	private int _port;
	private String _pattern;
	private int _scanBuffer;
	private int _expand;

	public RedisReader(LinkedBlockingQueue<String> queue) {
		this.queue = queue;
	}

	public void initialize() {
		
	}

	public void parseConfig() {
		XMLConfiguration config;
		try {
			config = new XMLConfiguration("conf.xml");
			this._host = config.getString("redis.host", "localhost1");
			this._port = config.getInt("redis.port", 5673);
			this._pattern = config.getString("redis.pattern", "logpumper_q");
			this._scanBuffer = config.getInt("redis.scanBuffer", 10000);
			this._expand = config.getInt("redis.expand", 1);
		} catch (ConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void run() {
		JedisPool pool = null;
		Jedis jedis = null;
		try {
			pool = new JedisPool(new JedisPoolConfig(), this._host, this._port, 0, null, 0);
			long now = System.currentTimeMillis();
			int counter = 0;
			jedis = pool.getResource();
			ScanParams scanParams = new ScanParams().match(this._pattern).count(this._scanBuffer);
			String cur = redis.clients.jedis.ScanParams.SCAN_POINTER_START;
			boolean cycleIsFinished = false;
			while (!cycleIsFinished) {

				ScanResult<String> scanResult = jedis.scan(cur, scanParams);
				List<String> keys = scanResult.getResult();
				String[] keys_a = keys.toArray(new String[keys.size()]);
				List<String> results = jedis.mget(keys_a);
				// flexjson.JSONDeserializer d = new
				// flexjson.JSONDeserializer();
				// flexjson.JSONSerializer s = new flexjson.JSONSerializer();
				for (String r : results) {
					for (int i = 0; i < this._expand; i++) {
						counter++;
						// HashMap<String, Object> map = (HashMap<String,
						// Object>)d.deserialize(r);
						// map.put("id", counter);
						// String rr = s.seri
						// HashMap map = (HashMap)d.deserialize(r);
						queue.offer(r);
					}
				}
				// do whatever with the key-value pairs in result

				cur = scanResult.getStringCursor();
				if (cur.equals("0")) {
					cycleIsFinished = true;
					queue.offer("_produced_end");
				}
			}
			long elapsed = System.currentTimeMillis() - now;
			LOG.info(String.format("emited %s records...", counter));
			LOG.info("reading from redis elapsed time " + elapsed);
		} finally {
			if (jedis != null)
				jedis.close();
			if (pool != null)
				pool.close();
		}
	}
}
